package cn.azzhu.day02.transformations;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 对DataStream进行操作，返回一个新的DataStream
 * @author azzhu
 * @create 2020-09-18 00:31:58
 */
public class TransformationDemo1 {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        final DataStreamSource<Integer> nums = env.fromElements(1, 2, 3, 4, 5);

        //map方法是一个transformation，功能：做映射
        final SingleOutputStreamOperator<Integer> res = nums.map(num -> num * 2).returns(Integer.class);

        //传入功能更加强大的RichMapFunction
        nums.map(new RichMapFunction<Integer, Integer>() {
            //构造方法之后，map方法之前，执行一次，Configuration 可以拿到全局配置
            //用来初始化一些连接，或者初始化或恢复state
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
            }

            @Override
            public Integer map(Integer value) throws Exception {
                return value * 2;
            }

            //销毁之前，执行一次，通常是做资源释放
            @Override
            public void close() throws Exception {
                super.close();
            }
        });
        //sink
        res.print();

        env.execute("TransformationDemo1");
    }
}
